package storm;

import java.io.BufferedReader;
import java.io.File;
import java.io.FileFilter;
import java.io.FileInputStream;
import java.io.IOException;
import java.io.InputStreamReader;
import java.util.Map;

import org.apache.storm.spout.SpoutOutputCollector;
import org.apache.storm.task.TopologyContext;
import org.apache.storm.topology.OutputFieldsDeclarer;
import org.apache.storm.topology.base.BaseRichSpout;
import org.apache.storm.tuple.Fields;
import org.apache.storm.tuple.Values;
import org.apache.storm.utils.Utils;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

/**
 * 
 * @author 李岩飞
 * @email eliyanfei@126.com
 * @date 2018年6月7日下午2:27:01
 */
public class LogSpout extends BaseRichSpout {
	public static Logger LOG = LoggerFactory.getLogger(LogSpout.class);
	SpoutOutputCollector _collector;
	String path = null;
	private boolean completed = false;

	public void open(Map conf, TopologyContext context, SpoutOutputCollector collector) {
		_collector = collector;
		path = "E:\\apache-tomcat-7.0.34\\logs";
	}

	@Override
	public void ack(Object msgId) {
		super.ack(msgId);
	}
	
	@Override
	public void fail(Object msgId) {
		super.fail(msgId);
	}
	
	public void nextTuple() {
		if (path == null) {
			Utils.sleep(100);
			return;
		}
		File file = new File(path);
		if (!file.exists()) {
			Utils.sleep(100);
			return;
		}
		if (file.listFiles().length == 0) {
			Utils.sleep(100);
			return;
		}
		file.listFiles(new FileFilter() {
			public boolean accept(File f) {
				if (f.isFile() && f.getName().startsWith("localhost_access_log")) {
					BufferedReader br = null;
					String line;
					try {
						br = new BufferedReader(new InputStreamReader(new FileInputStream(f)));
						while ((line = br.readLine()) != null) {
							String[] values = line.split(" ");
							LogBean logBean = new LogBean();
							logBean.ip = values[0];
							logBean.values = values;
							_collector.emit(new Values(logBean));
						}
					} catch (Exception e) {
					} finally {
						try {
							br.close();
						} catch (IOException e) {
							e.printStackTrace();
						}
					}
				}
				return false;
			}
		});
		completed = true;
	}

	public void declareOutputFields(OutputFieldsDeclarer declarer) {
		declarer.declare(new Fields("f1"));
	}
}